package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo5MapPartitions {
  // Map、MapPartitions
  // foreach、foreachPartitions
  // 有时在处理数据的时候需要和外部系统建立连接
  // 如果连接建立在Driver端 连接是不能被序列化的 及在算子内部无法使用算子外部创建的连接
  // 那么可以使用MapPartitions、foreachPartitions算子降低连接的创建与销毁次数，提高效率
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("Demo5MapPartitions")

    val sc: SparkContext = new SparkContext(conf)

    // 读取学生数据构建RDD
    val stuRDD: RDD[String] = sc.textFile("spark/data/students.txt", 6)
    println(stuRDD.getNumPartitions) // 这不是算子，只是RDD中的一个属性

    // take 也是一个action算子 会返回一个Array
    // 这里的foreach实际上是Array的方法 不是RDD的算子
    stuRDD.take(10).foreach(println)


    // 算子外的代码都是在Driver端运行
    // 算子内的代码是通过Driver分发到Executor中执行
    // 类似MySQL这种网络连接不能被序列化
    // 那就需要在算子内部去创建连接
    //    stuRDD.foreach(line => {
    //      // 相当于每处理一条数据就需要创建一次连接
    //      // 频繁的创建、关闭连接是非常耗时的
    //      // 1、加载驱动
    //      Class.forName("com.mysql.jdbc.Driver")
    //      // 2、创建连接
    //      val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student?useSSL=false", "root", "123456")
    //      // 3、创建Statement
    //      val stat: PreparedStatement = conn.prepareStatement("insert into student values(?,?,?,?,?)")
    //
    //      val splits: Array[String] = line.split(",")
    //      val id: Int = splits(0).toInt
    //      val name: String = splits(1)
    //      val age: Int = splits(2).toInt
    //      val gender: String = splits(3)
    //      val clazz: String = splits(4)
    //      stat.setInt(1, id)
    //      stat.setString(2, name)
    //      stat.setInt(3, age)
    //      stat.setString(4, gender)
    //      stat.setString(5, clazz)
    //      stat.execute()
    //      stat.close()
    //      conn.close()
    //    })


    // mapPartitions也是一个转换算子
    stuRDD.mapPartitions(rdd => {
      println("map partition")
      // 按分区去处理数据
      rdd.map(line => line.split(",")(1))
    }).foreach(println)

    // mapPartitionsWithIndex也是一个转换算子
    // 会在处理每一个分区的时候获得一个index
    stuRDD.mapPartitionsWithIndex((index, rdd) => {
      println("当前遍历的分区：" + index)
      // 按分区去处理数据
      rdd.map(line => line.split(",")(1))
    }).foreach(println)

    // foreachPartition也是一个操作算子
    // 每一个分区由一个task去处理数据
    // 相当于最终每个分区会创建一次conn
    stuRDD.foreachPartition(rdd => {
      // 1、加载驱动
      Class.forName("com.mysql.jdbc.Driver")
      // 2、创建连接
      val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student?useSSL=false", "root", "123456")
      println("创建了一次连接")
      // 3、创建Statement
      val stat: PreparedStatement = conn.prepareStatement("insert into student values(?,?,?,?,?)")

      println("map partition")
      // 按分区去处理数据
      rdd.foreach(line => {
        val splits: Array[String] = line.split(",")
        val id: Int = splits(0).toInt
        val name: String = splits(1)
        val age: Int = splits(2).toInt
        val gender: String = splits(3)
        val clazz: String = splits(4)
        stat.setInt(1, id)
        stat.setString(2, name)
        stat.setInt(3, age)
        stat.setString(4, gender)
        stat.setString(5, clazz)
        stat.addBatch()
      })
      // 使用批量插入
      stat.executeBatch()
      stat.close()
      conn.close()
    })


    while (true) {

    }
  }

}
